package com.g.ocache.impl.store;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.g.ocache.Cache;
import com.g.ocache.CacheValidator;
import com.g.ocache.Store;
import com.g.ocache.StoreEntry;
import com.g.ocache.event.StoreEventListener;
import com.g.ocache.event.StoreEventType;
import com.g.ocache.impl.store.expiration.Expirable;
import com.g.ocache.impl.store.expiration.ExpirationEvictor;


/**
 * 
 * @author Jingqi Xu
 */
public class ConcurrentStore<K, V> extends AbstractConcurrentStore<K, V> implements Expirable<K, V> {
	//
	private final int initialCapacity;
	private final ExpirationEvictor<K, V> evictor;
	private final StoreEntry.Factory<K, V> entryFactory;
	
	
	/**
	 * 
	 */
	public ConcurrentStore(Cache<K, V> cache) {
		this(cache, null, null, DEFAULT_INITIAL_CAPACITY, DEFAULT_CONSURRENCY_LEVEL, false);
	}
	
	public ConcurrentStore(Cache<K, V> cache, StoreEntry.Factory<K, V> sef) {
		this(cache, sef, null, DEFAULT_INITIAL_CAPACITY, DEFAULT_CONSURRENCY_LEVEL, false);
	}
	
	public ConcurrentStore(Cache<K, V> cache, StoreEntry.Factory<K, V> sef, ExpirationEvictor.Factory<K, V> ef) {
		this(cache, sef, ef, DEFAULT_INITIAL_CAPACITY, DEFAULT_CONSURRENCY_LEVEL, false);
	}
	
	public ConcurrentStore(Cache<K, V> cache, StoreEntry.Factory<K, V> sef,	ExpirationEvictor.Factory<K, V> ef, int initialCapacity, int concurrencyLevel, boolean fair) {
		//
		super(cache, concurrencyLevel, fair);
		
		//
		this.entryFactory = sef;
		this.initialCapacity = initialCapacity;
		this.evictor = (ef == null ? null : ef.create(this));
	}
	
	/**
	 * 
	 */
	protected void doStart() throws Exception {
		//
		super.doStart();
		
		//
		if(evictor != null) {
			evictor.start();
		}
	}
	
	protected void doStop() throws Exception {
		//
		if(evictor != null) {
			evictor.stop();
		}
		
		//
		super.doStop();
	}
	
	/**
	 * 
	 */
	public final int getInitialCapacity() {
		return initialCapacity;
	}
	
	public final ExpirationEvictor<K, V> getEvictor() {
		return evictor;
	}
	
	public final StoreEntry.Factory<K, V> getEntryFactory() {
		return entryFactory;
	}
	
	/**
	 * 
	 */
	@SuppressWarnings("unchecked")
	public Map<K, V> detect() {
		//
		Map<K, V> expired = new HashMap<K, V>();
		if(entryFactory == null) {
			return expired;
		}
		
		//
		lock(true); // Lock all buckets
		try {
			for(Bucket<K, V> bucket : getBuckets()) {
				if(bucket instanceof Expirable) {
					expired.putAll(((Expirable<K, V>)bucket).detect());
				}
			}
		} finally {
			unlock(true);
		}
		
		//
		for(Entry<K, V> entry : expired.entrySet()) {
			super.publish(StoreEventType.expire, entry.getKey(), entry.getValue());
		}
		return expired;
	}

	/**
	 * 
	 */
	protected List<Bucket<K, V>> createBuckets(int count) {
		List<Bucket<K, V>> r = new ArrayList<Bucket<K, V>>(count);
		for(int i = 0; i < count; i++) {
			if(entryFactory == null) {
				r.add(new BasicBucket(getCache(), isFair(), getInitialCapacity()));
			} else {
				r.add(new DefaultBucket(getCache(), isFair(), getInitialCapacity(), entryFactory));
			}
		}
		return r;
	}
	
	/**
	 * 
	 */
	private class BasicBucket implements Bucket<K, V> {
		//
		private final Cache<K, V> cache;
		private final HashMap<K, V> values;
		private final ReentrantReadWriteLock lock;
		
		/**
		 * 
		 */
		public BasicBucket(Cache<K, V> cache, boolean fair, int initialCapacity) {
			this.cache = cache;
			this.lock = new ReentrantReadWriteLock(fair);
			this.values = new HashMap<K, V>(initialCapacity);
		}
		
		/**
		 * 
		 */
		private void lock(boolean exclusive) {
			final Lock lock = exclusive ? this.lock.writeLock() : this.lock.readLock();
			lock.lock();
		}
		
		private void unlock(boolean exclusive) {
			final Lock lock = exclusive ? this.lock.writeLock() : this.lock.readLock();
			lock.unlock();
		}

		/**
		 * 
		 */
		public int size() {
			return values.size();
		}
		
		public boolean containsKey(K key) {
			lock(false);
			try {
				return values.containsKey(key);
			} finally {
				unlock(false);
			}
		}
		
		public void clear() {
			values.clear();
		}

		public V remove(K key) {
			lock(true);
			try {
				return values.remove(key);
			} finally {
				unlock(true);
			}
		}
		
		public V get(K key) {
			lock(false);
			try {
				return values.get(key);
			} finally {
				unlock(false);
			}
		}
		
		public Map<K, V> getAll() {
			return values;
		}
		
		public V put(K key, V value) {
			lock(true);
			try {
				//
				final CacheValidator<K, V> validator = cache.getValidator();
				if(validator != null) {
					V existing = get(key);
					validator.validate(key, existing, value);
				}
				
				//
				return values.put(key, value);
			} finally {
				unlock(true);
			}
		}
		
		public V putIfAbsent(K key, V value) {
			lock(true);
			try {
				//
				final CacheValidator<K, V> validator = cache.getValidator();
				if(validator != null) {
					validator.validate(key, null, value);
				}

				//
				V existing = values.get(key);
				if(existing == null) {
					values.put(key, value);
				}
				return existing;
			} finally {
				unlock(true);
			}
		}
	}
	
	private class DefaultBucket implements Bucket<K, V>, Expirable<K, V> {
		//
		private final Cache<K, V> cache;
		private final ReentrantReadWriteLock lock;
		private final HashMap<K, StoreEntry<K, V>> values;
		private final StoreEntry.Factory<K, V> entryFactory;
		
		/**
		 * 
		 */
		public DefaultBucket(Cache<K, V> cache, boolean fair, int initialCapacity, StoreEntry.Factory<K, V> sef) {
			this.cache = cache;
			this.entryFactory = sef;
			this.lock = new ReentrantReadWriteLock(fair);
			this.values = new HashMap<K, StoreEntry<K, V>>(initialCapacity);
		}
		
		/**
		 * 
		 */
		private void lock(boolean exclusive) {
			final Lock lock = exclusive ? this.lock.writeLock() : this.lock.readLock();
			lock.lock();
		}
		
		private void unlock(boolean exclusive) {
			final Lock lock = exclusive ? this.lock.writeLock() : this.lock.readLock();
			lock.unlock();
		}

		/**
		 * 
		 */
		public int size() { // Guarded by AbstractConcurrentStore
			return values.size();
		}
		
		public boolean containsKey(K key) {
			//
			StoreEntry<K, V> existing = null;
			lock(false);
			try {
				existing = this.values.get(key);
			} finally {
				unlock(false);
			}
			
			//
			return existing == null || existing.isExpired() ? false : true;
		}
		
		public void clear() { // Guarded by AbstractConcurrentStore
			this.values.clear();
		}

		public V remove(K key) {
			//
			StoreEntry<K, V> existing = null;
			lock(true);
			try {
				existing = this.values.remove(key);
			} finally {
				unlock(true);
			}
			
			//
			if(existing != null && existing.isExpired()) {
				publish(StoreEventType.expire, existing);
			}
			return existing == null || existing.isExpired() ? null : existing.getValue();
		}
		
		public V get(K key) {
			//
			StoreEntry<K, V> existing = null;
			lock(false);
			try {
				existing = this.values.get(key);
			} finally {
				unlock(false);
			}
			
			//
			return existing == null || existing.isExpired() ? null : existing.getValue();
		}
		
		public Map<K, V> getAll() { // Guarded by AbstractConcurrentStore
			Map<K, V> r = new HashMap<K, V>(this.values.size());
			for(K key : this.values.keySet()) {
				StoreEntry<K, V> entry = this.values.get(key);
				if (entry != null && !entry.isExpired()) {
					r.put(key, entry.getValue());
				}
			}
			return r;
		}
		
		public V put(K key, V value) {
			//
			StoreEntry<K, V> existing = null;
			lock(true);
			try {
				//
				final CacheValidator<K, V> validator = cache.getValidator();
				if(validator != null) {
					existing = this.values.get(key);
					V v = existing == null || existing.isExpired() ? null : existing.getValue();
					validator.validate(key, v, value);
				}
				
				//
				existing = this.values.put(key, entryFactory.create(key, value));
			} finally {
				unlock(true);
			}
			
			//
			if(existing != null && existing.isExpired()) {
				publish(StoreEventType.expire, existing);
			}
			return existing == null || existing.isExpired() ? null : existing.getValue();
		}
		
		public V putIfAbsent(K key, V value) {
			//
			StoreEntry<K, V> existing = null;
			lock(true);
			try {
				//
				existing = this.values.get(key);
				if(existing == null || existing.isExpired()) {
					//
					final CacheValidator<K, V> validator = cache.getValidator();
					if(validator != null) {
						validator.validate(key, null, value);
					}
					
					//
					this.values.put(key, entryFactory.create(key, value));
				}
			} finally {
				unlock(true);
			}
			
			//
			if(existing != null && existing.isExpired()) {
				publish(StoreEventType.expire, existing);
			}
			return existing == null || existing.isExpired() ? null : existing.getValue();
		}

		/**
		 * 
		 */
		public Map<K, V> detect() {
			Map<K, V> expired = new HashMap<K, V>();
			lock(true);
			try {
				for(Iterator<K> iter = values.keySet().iterator(); iter.hasNext(); ) {
					K key = iter.next();
					
					//
					StoreEntry<K, V> entry = values.get(key);
					if(entry != null && entry.isExpired()) {
						//
						expired.put(key, entry.getValue());
						
						//
						iter.remove();
					}
				}
			} finally {
				unlock(true);
			}
			
			//
			for(Entry<K, V> e : expired.entrySet()) {
				publish(StoreEventType.expire, e.getKey(), e.getValue());
			}
			return expired;
		}
	}

	/**
	 * 
	 */
	public static class FactoryImpl<K, V> implements Factory<K, V> {
		//
		private boolean fair;
		private StoreEventListener<K, V> listener;
		private StoreEntry.Factory<K, V> entryFactory;
		private ExpirationEvictor.Factory<K, V> evictorFactory;
		private int initialCapacity = AbstractConcurrentStore.DEFAULT_INITIAL_CAPACITY;
		private int concurrencyLevel = AbstractConcurrentStore.DEFAULT_CONSURRENCY_LEVEL;
		
		/**
		 * 
		 */
		public FactoryImpl() {
			this(null);
		}
		
		public FactoryImpl(StoreEntry.Factory<K, V> sef) {
			this.entryFactory = sef;
		}
		
		/**
		 * 
		 */
		public Store<K, V> create(Cache<K, V> cache) {
			Store<K, V> r = new ConcurrentStore<K, V>(cache, entryFactory, evictorFactory, initialCapacity, concurrencyLevel, fair);
			r.addStoreEventListener(getListener());
			return r;
		}

		/**
		 * 
		 */
		public boolean isFair() {
			return fair;
		}

		public void setFair(boolean fair) {
			this.fair = fair;
		}
		
		public int getInitialCapacity() {
			return initialCapacity;
		}

		public void setInitialCapacity(int ic) {
			this.initialCapacity = ic;
		}

		public int getConcurrencyLevel() {
			return concurrencyLevel;
		}

		public void setConcurrencyLevel(int cl) {
			this.concurrencyLevel = cl;
		}

		public StoreEventListener<K, V> getListener() {
			return listener;
		}

		public void setListener(StoreEventListener<K, V> listener) {
			this.listener = listener;
		}
		
		public StoreEntry.Factory<K, V> getEntryFactory() {
			return entryFactory;
		}

		public void setEntryFactory(StoreEntry.Factory<K, V> entryFactory) {
			this.entryFactory = entryFactory;
		}
		
		public ExpirationEvictor.Factory<K, V> getEvictorFactory() {
			return evictorFactory;
		}

		public void setEvictorFactory(ExpirationEvictor.Factory<K, V> evictorFactory) {
			this.evictorFactory = evictorFactory;
		}
	}
}
